package com.example.demo.demos.service.impl;

import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.io.*;
import java.math.BigDecimal;
import java.sql.*;
import java.util.*;

/**
 *
 * 本地文件测试 可 配置表映射与主键映射
 */
public class LocalGzDataSyncTest_bk {

    // 配置常量
    private static final String LOCAL_ROOT_PATH = "D:/test_data/";
    private static final String DB_URL = "jdbc:highgo://localhost:5866/highgo?currentSchema=public";
    private static final String DB_USER = "highgo";
    private static final String DB_PASS = "root";
    private static final int BATCH_SIZE = 1000;

    // 表映射配置
    private static final Map<String, String> TABLE_MAPPING = new HashMap<>();
    private static final Map<String, String> TABLE_PK_FIELDS = new HashMap<>();
    private static final Map<String, Map<String, Integer>> TABLE_SCHEMA_CACHE = new HashMap<>();

    static {
        // 配置表映射
        TABLE_MAPPING.put("company_chattel", "company_chattel");

        // 配置主键字段
        TABLE_PK_FIELDS.put("company_chattel", "id");
    }

    public static void main(String[] args) {
        try {
            syncAllTables();
            System.out.println("所有表数据同步完成");
        } catch (Exception e) {
            System.err.println("数据同步失败: " + e.getMessage());
            e.printStackTrace();
        }
    }

    public static void syncAllTables() throws IOException, SQLException {
        File rootDir = new File(LOCAL_ROOT_PATH);
        validateDirectory(rootDir);

        File[] tableDirs = rootDir.listFiles(File::isDirectory);
        if (tableDirs != null) {
            try (Connection connection = DriverManager.getConnection(DB_URL, DB_USER, DB_PASS)) {
                connection.setAutoCommit(false);

                try {
                    for (File tableDir : tableDirs) {
                        processTableDirectory(connection, tableDir);
                    }
                    connection.commit();
                } catch (Exception e) {
                    connection.rollback();
                    throw new SQLException("事务回滚: " + e.getMessage(), e);
                }
            }
        }
    }

    private static void validateDirectory(File dir) throws IOException {
        if (!dir.exists() || !dir.isDirectory()) {
            throw new IOException("目录不存在或不是有效目录: " + dir.getAbsolutePath());
        }
    }

    private static void processTableDirectory(Connection conn, File tableDir) throws SQLException, IOException {
        String tableDirName = tableDir.getName();
        String tableName = TABLE_MAPPING.get(tableDirName);

        if (tableName == null) {
            System.out.println("跳过未映射的目录: " + tableDirName);
            return;
        }

        System.out.println("开始处理表: " + tableName);
        loadTableSchema(conn, tableName);

        File[] dateDirs = tableDir.listFiles(File::isDirectory);
        if (dateDirs != null) {
            for (File dateDir : dateDirs) {
                processDateDirectory(conn, tableName, dateDir);
            }
        }
    }

    private static void loadTableSchema(Connection conn, String tableName) throws SQLException {
        if (!TABLE_SCHEMA_CACHE.containsKey(tableName)) {
            Map<String, Integer> columnTypes = new HashMap<>();
            DatabaseMetaData metaData = conn.getMetaData();

            try (ResultSet rs = metaData.getColumns(null, null, tableName, null)) {
                while (rs.next()) {
                    columnTypes.put(rs.getString("COLUMN_NAME"), rs.getInt("DATA_TYPE"));
                }
            }

            TABLE_SCHEMA_CACHE.put(tableName, columnTypes);
            System.out.println("已加载表结构: " + tableName + ", 列数: " + columnTypes.size());
        }
    }

    private static void processDateDirectory(Connection conn, String tableName, File dateDir) throws IOException, SQLException {
        System.out.println("处理日期目录: " + dateDir.getName());

        File[] gzFiles = dateDir.listFiles((dir, name) -> name.endsWith(".gz"));
        if (gzFiles != null) {
            for (File gzFile : gzFiles) {
                processGzFile(conn, tableName, gzFile);
            }
        }
    }

    private static void processGzFile(Connection conn, String tableName, File gzFile) throws IOException, SQLException {
        System.out.println("处理文件: " + gzFile.getName());

        ObjectMapper objectMapper = new ObjectMapper();
        StringBuilder jsonContent = new StringBuilder("[");

        try (FileInputStream fis = new FileInputStream(gzFile);
             GzipCompressorInputStream gzIn = new GzipCompressorInputStream(fis);
             BufferedReader reader = new BufferedReader(new InputStreamReader(gzIn, "UTF-8"))) {

            String line;
            while ((line = reader.readLine()) != null) {
                jsonContent.append(line).append(",");
            }

            if (jsonContent.length() > 1) {
                jsonContent.setLength(jsonContent.length() - 1);
            }
            jsonContent.append("]");

            JsonNode rootNode = objectMapper.readTree(jsonContent.toString());
            processJsonData(conn, tableName, rootNode);
        }
    }

    private static void processJsonData(Connection conn, String tableName, JsonNode rootNode) throws SQLException {
        if (!rootNode.isArray()) {
            throw new IllegalArgumentException("JSON数据必须是数组格式");
        }

        String pkField = TABLE_PK_FIELDS.get(tableName);
        if (pkField == null) {
            throw new SQLException("表" + tableName + "未配置主键字段");
        }

        Map<String, Integer> columnTypes = TABLE_SCHEMA_CACHE.get(tableName);
        if (columnTypes == null) {
            throw new SQLException("无法获取表" + tableName + "的结构信息");
        }

        // 不去重，直接处理所有记录
        List<Map<String, Object>> records = new ArrayList<>();
        for (JsonNode node : rootNode) {
            Map<String, Object> fieldMap = parseJsonNode(node, columnTypes);
            Object pkValue = fieldMap.get(pkField);

            if (pkValue == null) {
                System.err.println("警告: 跳过缺少主键的记录");
                continue;
            }

            records.add(fieldMap);
        }

        System.out.println("待处理记录数: " + records.size());

        // 根据数据库类型选择处理方式
        if (isPostgreSQL(conn)) {
            executePostgresUpsert(conn, tableName, pkField, records);
        } else {
            executeGenericUpsert(conn, tableName, pkField, records);
        }
    }

    private static Map<String, Object> parseJsonNode(JsonNode node, Map<String, Integer> columnTypes) {
        Map<String, Object> fieldMap = new HashMap<>();
        Iterator<String> fieldNames = node.fieldNames();

        while (fieldNames.hasNext()) {
            String fieldName = fieldNames.next();
            if (columnTypes.containsKey(fieldName)) {
                JsonNode valueNode = node.get(fieldName);
                Object value = convertJsonValue(valueNode, columnTypes.get(fieldName));
                fieldMap.put(fieldName, value);
            }
        }

        return fieldMap;
    }

    private static Object convertJsonValue(JsonNode node, int sqlType) {
        if (node == null || node.isNull()) {
            return null;
        }

        try {
            switch (sqlType) {
                case Types.SMALLINT:
                    return node.canConvertToInt() ? node.asInt() : null;
                case Types.INTEGER:
                    return node.asInt();
                case Types.BIGINT:
                    return node.asLong();
                case Types.DECIMAL:
                case Types.NUMERIC:
                    return node.isNumber() ? node.decimalValue() : new BigDecimal(node.asText());
                case Types.FLOAT:
                case Types.DOUBLE:
                    return node.asDouble();
                case Types.BOOLEAN:
                case Types.BIT:
                    if (node.isTextual()) {
                        String text = node.asText().toLowerCase();
                        if ("true".equals(text) || "1".equals(text)) return true;
                        if ("false".equals(text) || "0".equals(text)) return false;
                    }
                    return node.asBoolean();
                case Types.DATE:
                    return parseDate(node.asText());
                case Types.TIMESTAMP:
                    return parseTimestamp(node.asText());
                case Types.TIME:
                    return parseTime(node.asText());
                default:
                    return node.asText();
            }
        } catch (IllegalArgumentException e) {
            System.err.println("类型转换失败: " + e.getMessage());
            return null;
        }
    }

    private static java.sql.Date parseDate(String dateStr) {
        try {
            // 去掉可能存在的时区信息
            String cleanDate = dateStr.split(" ")[0];
            return java.sql.Date.valueOf(cleanDate);
        } catch (Exception e) {
            System.err.println("日期格式错误: " + dateStr);
            return null;
        }
    }

    private static Timestamp parseTimestamp(String timestampStr) {
        try {
            // 处理可能存在的多种时间格式
            String cleanTimestamp = timestampStr.trim();

            // 如果包含时区信息，只取前面的部分
            if (cleanTimestamp.contains(" ")) {
                cleanTimestamp = cleanTimestamp.split(" ")[0];
            }

            // 确保格式为"yyyy-MM-dd HH:mm:ss"
            if (cleanTimestamp.length() == 10) { // 只有日期部分
                cleanTimestamp += " 00:00:00";
            } else if (cleanTimestamp.length() == 16) { // 缺少秒数
                cleanTimestamp += ":00";
            }

            return Timestamp.valueOf(cleanTimestamp);
        } catch (Exception e) {
            System.err.println("时间戳格式错误: " + timestampStr + ", 错误: " + e.getMessage());
            return null;
        }
    }

    private static Time parseTime(String timeStr) {
        try {
            // 去掉可能存在的时区信息
            String cleanTime = timeStr.split(" ")[0];
            return Time.valueOf(cleanTime);
        } catch (Exception e) {
            System.err.println("时间格式错误: " + timeStr);
            return null;
        }
    }

    private static boolean isPostgreSQL(Connection conn) throws SQLException {
        return conn.getMetaData().getDatabaseProductName().equals("PostgreSQL");
    }

    private static void executePostgresUpsert(Connection conn, String tableName, String pkField,
                                              List<Map<String, Object>> records) throws SQLException {
        if (records.isEmpty()) {
            System.out.println("没有需要处理的记录");
            return;
        }

        // 获取表结构信息
        Map<String, Integer> columnTypes = TABLE_SCHEMA_CACHE.get(tableName);

        // 准备字段列表
        Set<String> fields = records.get(0).keySet();
        String[] fieldArray = fields.toArray(new String[0]);

        // 构建UPSERT语句
        StringBuilder sql = new StringBuilder("INSERT INTO ")
                .append(tableName).append(" (")
                .append(String.join(",", fieldArray)).append(") VALUES (")
                .append(String.join(",", Collections.nCopies(fieldArray.length, "?")))
                .append(") ON CONFLICT (").append(pkField).append(") DO UPDATE SET ");

        // 构建UPDATE部分
        for (String field : fieldArray) {
            if (!field.equals(pkField)) {
                sql.append(field).append(" = EXCLUDED.").append(field).append(",");
            }
        }
        sql.setLength(sql.length() - 1);

        System.out.println("执行SQL: " + sql.toString());

        // 执行批量操作
        try (PreparedStatement stmt = conn.prepareStatement(sql.toString())) {
            int count = 0;
            int successCount = 0;

            for (Map<String, Object> record : records) {
                try {
                    for (int i = 0; i < fieldArray.length; i++) {
                        String fieldName = fieldArray[i];
                        Object value = record.get(fieldName);
                        setStatementParameter(stmt, i + 1, value, columnTypes.get(fieldName));
                    }
                    stmt.addBatch();
                    successCount++;

                    if (++count % BATCH_SIZE == 0) {
                        stmt.executeBatch();
                        System.out.println("已处理 " + count + " 条记录");
                    }
                } catch (SQLException e) {
                    System.err.println("处理记录失败: " + e.getMessage());
                }
            }

            // 执行剩余批次
            int[] results = stmt.executeBatch();
            System.out.println("完成处理，共 " + count + " 条记录，成功 " + successCount + " 条");
        }
    }

    private static void executeGenericUpsert(Connection conn, String tableName, String pkField,
                                             List<Map<String, Object>> records) throws SQLException {
        if (records.isEmpty()) {
            System.out.println("没有需要处理的记录");
            return;
        }

        // 获取表结构信息
        Map<String, Integer> columnTypes = TABLE_SCHEMA_CACHE.get(tableName);

        // 准备字段列表
        Set<String> fields = records.get(0).keySet();
        String[] fieldArray = fields.toArray(new String[0]);

        // 准备SQL语句
        String checkSql = String.format("SELECT 1 FROM %s WHERE %s = ?", tableName, pkField);
        String insertSql = buildInsertSql(tableName, fieldArray);
        String updateSql = buildUpdateSql(tableName, pkField, fieldArray);

        System.out.println("检查SQL: " + checkSql);
        System.out.println("插入SQL: " + insertSql);
        System.out.println("更新SQL: " + updateSql);

        try (PreparedStatement checkStmt = conn.prepareStatement(checkSql);
             PreparedStatement insertStmt = conn.prepareStatement(insertSql);
             PreparedStatement updateStmt = conn.prepareStatement(updateSql)) {

            int insertCount = 0;
            int updateCount = 0;
            int errorCount = 0;

            for (Map<String, Object> record : records) {
                try {
                    Object pkValue = record.get(pkField);

                    // 检查记录是否存在
                    checkStmt.clearParameters();
                    setStatementParameter(checkStmt, 1, pkValue, columnTypes.get(pkField));

                    boolean exists = checkStmt.executeQuery().next();

                    if (exists) {
                        // 执行更新
                        int paramIndex = 1;
                        for (String field : fieldArray) {
                            if (!field.equals(pkField)) {
                                setStatementParameter(updateStmt, paramIndex++, record.get(field), columnTypes.get(field));
                            }
                        }
                        setStatementParameter(updateStmt, paramIndex, pkValue, columnTypes.get(pkField));
                        updateStmt.addBatch();
                        updateCount++;

                        if (updateCount % BATCH_SIZE == 0) {
                            updateStmt.executeBatch();
                        }
                    } else {
                        // 执行插入
                        int paramIndex = 1;
                        for (String field : fieldArray) {
                            setStatementParameter(insertStmt, paramIndex++, record.get(field), columnTypes.get(field));
                        }
                        insertStmt.addBatch();
                        insertCount++;

                        if (insertCount % BATCH_SIZE == 0) {
                            insertStmt.executeBatch();
                        }
                    }
                } catch (Exception e) {
                    System.err.println("处理记录失败: " + e.getMessage());
                    errorCount++;
                }
            }

            // 执行剩余批次
            if (insertCount > 0) {
                insertStmt.executeBatch();
            }
            if (updateCount > 0) {
                updateStmt.executeBatch();
            }

            System.out.printf("完成处理: 新增 %d 条, 更新 %d 条, 失败 %d 条%n", insertCount, updateCount, errorCount);
        }
    }

    private static String buildInsertSql(String tableName, String[] fields) {
        return String.format("INSERT INTO %s (%s) VALUES (%s)",
                tableName,
                String.join(",", fields),
                String.join(",", Collections.nCopies(fields.length, "?")));
    }

    private static String buildUpdateSql(String tableName, String pkField, String[] fields) {
        StringBuilder sql = new StringBuilder("UPDATE ").append(tableName).append(" SET ");

        for (String field : fields) {
            if (!field.equals(pkField)) {
                sql.append(field).append(" = ?,");
            }
        }
        sql.setLength(sql.length() - 1);
        sql.append(" WHERE ").append(pkField).append(" = ?");

        return sql.toString();
    }

    private static void setStatementParameter(PreparedStatement stmt, int index, Object value, int sqlType)
            throws SQLException {
        if (value == null) {
            stmt.setNull(index, sqlType);
        } else {
            try {
                switch (sqlType) {
                    case Types.SMALLINT:
                        if (value instanceof Number) {
                            stmt.setShort(index, ((Number)value).shortValue());
                        } else if (value instanceof Boolean) {
                            stmt.setShort(index, (short)(((Boolean)value) ? 1 : 0));
                        } else {
                            stmt.setShort(index, Short.parseShort(value.toString()));
                        }
                        break;
                    case Types.INTEGER:
                        stmt.setInt(index, value instanceof Number ? ((Number)value).intValue() : Integer.parseInt(value.toString()));
                        break;
                    case Types.BIGINT:
                        stmt.setLong(index, value instanceof Number ? ((Number)value).longValue() : Long.parseLong(value.toString()));
                        break;
                    case Types.DECIMAL:
                    case Types.NUMERIC:
                        stmt.setBigDecimal(index, value instanceof BigDecimal ? (BigDecimal)value : new BigDecimal(value.toString()));
                        break;
                    case Types.FLOAT:
                    case Types.DOUBLE:
                        stmt.setDouble(index, value instanceof Number ? ((Number)value).doubleValue() : Double.parseDouble(value.toString()));
                        break;
                    case Types.BOOLEAN:
                    case Types.BIT:
                        stmt.setBoolean(index, value instanceof Boolean ? (Boolean)value : Boolean.parseBoolean(value.toString()));
                        break;
                    case Types.DATE:
                        if (value instanceof java.sql.Date) {
                            stmt.setDate(index, (java.sql.Date)value);
                        } else if (value instanceof java.util.Date) {
                            stmt.setDate(index, new java.sql.Date(((java.util.Date)value).getTime()));
                        } else {
                            stmt.setDate(index, java.sql.Date.valueOf(value.toString()));
                        }
                        break;
                    case Types.TIMESTAMP:
                        if (value instanceof Timestamp) {
                            stmt.setTimestamp(index, (Timestamp)value);
                        } else if (value instanceof java.util.Date) {
                            stmt.setTimestamp(index, new Timestamp(((java.util.Date)value).getTime()));
                        } else {
                            stmt.setTimestamp(index, Timestamp.valueOf(value.toString()));
                        }
                        break;
                    case Types.TIME:
                        if (value instanceof Time) {
                            stmt.setTime(index, (Time)value);
                        } else {
                            stmt.setTime(index, Time.valueOf(value.toString()));
                        }
                        break;
                    default:
                        stmt.setObject(index, value);
                }
            } catch (Exception e) {
                System.err.println("设置参数失败: " + e.getMessage());
                stmt.setNull(index, sqlType);
            }
        }
    }
}